package com.qupeng.demo.kafka.kafkaspringbootstreams.timewindow;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.utils.TimeUtils;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.ProcessorContextImpl;

public class WindowedLongProcessor implements Processor<Windowed<String>, Long, Object, Object> {

    private ProcessorContextImpl context;

    @Override
    public void init(ProcessorContext<Object, Object> context) {
        Processor.super.init(context);
        this.context = (ProcessorContextImpl)context;
    }

    @Override
    public void process(Record<Windowed<String>, Long> record) {
        System.out.println("Current stream time is: " + TimeUtils.format(this.context.currentStreamTimeMs()));
        System.out.println("Current system time is: " + TimeUtils.format(this.context.currentSystemTimeMs()));
        System.out.println("Time window is from : "
                + TimeUtils.format(record.key().window().start())
                + " to "
                + TimeUtils.format(record.key().window().end()));
    }

    @Override
    public void close() {
        Processor.super.close();
    }
}